package org.nbict.iot.protocol.trident;

import org.apache.storm.trident.operation.Filter;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.tuple.TridentTuple;

import java.util.Map;

/**
 * Created by songseven on 18/6/29.
 */
public class DisplayOutputFilter implements Filter {
    private int partitionIndex;

    @Override
    public void prepare(Map conf, TridentOperationContext context) {
        this.partitionIndex = context.getPartitionIndex();
    }
    @Override
    public void cleanup() {
    }
    @Override
    public boolean isKeep(TridentTuple tuple) {

//        System.out.print("I am partition [" + partitionIndex+ "] : [");
//        for (int index = 0 ; index < tuple.size(); index ++){
//            System.out.print(tuple.get(index));
//            if(index < (tuple.size()-1))
//                System.out.print(",");
//        }
//        System.out.println("]");

//        Utils.sleep(1000);

        System.out.println(partitionIndex + ": " + tuple.get(0) + " => " + tuple.get(1));

        return true;
    }
}
